package ooo.reindeer.concurrent;

import ooo.reindeer.concurrent.filequeue.FSQueue;
import ooo.reindeer.concurrent.filequeue.exception.FileFormatException;
import ooo.reindeer.logging.ILogger;
import ooo.reindeer.logging.LogUtil;

import java.io.File;
import java.io.IOException;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


public class FileQueue<T> extends AbstractQueue<T> implements Queue<T>,
        java.io.Serializable {
    private static final ILogger logger = LogUtil.getLogger(FileQueue.class);
    private static final long serialVersionUID = -5960741434564940154L;
    protected ReentrantReadWriteLock locker = new ReentrantReadWriteLock();
    protected Lock wlock = locker.writeLock();
    private FSQueue fsQueue = null;
    //	private Condition blocker = rlock.newCondition();
    private String rootPath = null;

    private ICoder<T, byte[]> coder;

    public FileQueue(String path, ICoder<T, byte[]> coder) throws Exception {
        this(path, 1024 * 1024 * 16, coder);
    }

    public FileQueue(String path, int logsize, ICoder<T, byte[]> coder) throws Exception {
        rootPath = path;
        fsQueue = new FSQueue(path, logsize);
        this.coder = coder;
    }

    public void close() {
        if (fsQueue != null) {
            try {
                wlock.lock();
                fsQueue.close();
            }finally {
                wlock.unlock();
            }
        }
    }

    public void destroy() {
        this.close();
        File file = new File(rootPath);

        File[] dbFiles = file.listFiles();
        for (File f : dbFiles) {
            if (f.isFile() && (f.getName().endsWith(".db") || f.getName().endsWith(".idb"))) {
                System.out.println(f.getName());
                System.out.println("delete = " + f.delete());
            }
        }

    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof FileQueue)) {
            return false;
        }
        FileQueue bytes = (FileQueue) o;
        return locker.equals(bytes.locker) &&
                rootPath.equals(bytes.rootPath);
    }

    @Override
    public int hashCode() {
        return rootPath.hashCode();
    }

    @Override
    public Iterator<T> iterator() {
        throw new UnsupportedOperationException("iterator Unsupported now");
    }

    @Override
    public boolean offer(T e) {

        byte[] data = coder.encode(e);


        if (fsQueue.getQueuSize() == Integer.MAX_VALUE) {
            return false;
        } else {
            try {
                wlock.lock();

                fsQueue.add(data);

                return true;
            } catch (Exception ex) {
                throw new RuntimeException(ex);
            } finally {
                wlock.unlock();
            }
        }


    }

    @Override
    public T peek() {
        throw new UnsupportedOperationException("peek Unsupported now");
    }

    @Override
    public T poll() {
        byte[] temp = null;
        try {
            wlock.lock();
            temp = fsQueue.readNextAndRemove();

        } catch (IOException e) {
            logger.error(e.getMessage(), e);
            return null;
        } catch (FileFormatException e) {
            logger.error(e.getMessage(), e);
            return null;
        } finally {
            wlock.unlock();
        }
        if (temp == null) {
            return null;
        } else {

            return coder.decode(temp);

        }
    }

    @Override
    public int size() {
        return fsQueue.getQueuSize();
    }
}
